package cn.acewill.lrz.activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**

 */
public class JmsReceiver {
    public static void main(String[] args) throws Exception {
        JmsReceiver jms = new JmsReceiver();
//        jms.queueReceiver();
        jms.topicReceiver();
    }

    /**
     * 话题消息接收
     * 一定要先运行一次， 等于向消息服务中间件注册这个消费者， 然后再运行客户端发送信息， 这个时候， 无
     论消费者是否在线， 都会接收到， 不在线的话， 下次连接的时候， 会把没有收过的消息都接收下来。
     * @throws Exception
     */
    private void topicReceiver() throws Exception{
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        Connection connection = cf.createConnection();
        //需要在连接上设置消费者id， 用来识别消费者
        connection.setClientID("join");
        final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Topic destination = session.createTopic("MyTopic");
        //需要创建TopicSubscriber来订阅
        TopicSubscriber ts = session.createDurableSubscriber(destination, "T1");
        //要设置好了过后再start 这个 connection
        connection.start();
        Message message = ts.receive();
        while(message!=null) {
            TextMessage txtMsg = (TextMessage)message; session.commit();
            System.out.println("收到消息： " + txtMsg.getText());
            message = ts.receive(1000L);
        }
        session.close();
        connection.close();
    }

    /**
     * Queue消息接收
     * @throws Exception
     */
    private void queueReceiver() throws Exception{
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        Connection connection = cf.createConnection();
        connection.start();
        final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("my-queue");
        //JMS consumer： 消息消费者， 接收和处理JMS消息的客户端应用消息的消费,可以采用以下两种方法之一：
        //1： 同步消费： 通过调用消费者的receive方法从目的地中显式提取消息， receive 方法可以一直阻塞到消息到达。
        //2： 异步消费： 客户可以为消费者注册一个消息监听器， 以定义在消息到达时所采取的动作
        MessageConsumer consumer = session.createConsumer(destination);
        int i=0;
        while(i<3) {
            i++;
            TextMessage message = (TextMessage) consumer.receive();
            session.commit();
            System.out.println("收到消息： " + message.getText());
        }
        session.close();
        connection.close();
    }
}
